package mq;

import java.io.UnsupportedEncodingException;
import java.util.List;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;


/**
 * 
 * @author zyb
 *
 * 2018年2月21日 下午12:29:11
 */
public class RpcServer {
	
	private DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("rpcserver");
	
	private Serializer serializer = new ProtostuffSerializer();
	
	public void start() {
		consumer.setNamesrvAddr("127.0.0.1:9876");
	}

	public void subscribe() {
		try {
			consumer.subscribe("fuck2", "*");
		} catch (MQClientException e) {
			e.printStackTrace();
		}
		consumer.registerMessageListener(new MessageListenerOrderly() {
			
			@Override
			public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
				for (MessageExt msg : msgs) {
//					System.out.println(serializer.deserialize(msg.getBody(), Request.class));
					try {
						System.out.println(new String(msg.getBody(), "utf-8"));
					} catch (UnsupportedEncodingException e) {
						e.printStackTrace();
					}
				}
				return ConsumeOrderlyStatus.SUCCESS;
			}
		});
		try {
			consumer.start();
		} catch (MQClientException e) {
			e.printStackTrace();
		}
	}
	
	public static void main(String[] args) {
		RpcServer server = new RpcServer();
		server.start();
		server.subscribe();
	}
	
}
